/*-------------------------------------------------------------------------
 *
 * test_decoding.c
 *		  示例逻辑解码输出插件
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		  contrib/test_decoding/test_decoding.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "catalog/pg_type.h"

#include "replication/logical.h"
#include "replication/origin.h"

#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"

PG_MODULE_MAGIC;

/* 这些必须对 dlsym() 可用 */
extern void _PG_init(void);
extern void _PG_output_plugin_init(OutputPluginCallbacks *fc_cb);

typedef struct
{
	MemoryContext context;
	bool		include_xids;
	bool		include_timestamp;
	bool		skip_empty_xacts;
	bool		only_local;
} TestDecodingData;

/*
 * 维护每个事务级别的变量，以跟踪事务和/或流是否已写入任何更改。 在流模式下，事务可以在流中解码，因此除了保持事务是否已写入任何更改外，我们还需要跟踪当前流是否已写入任何更改。 这是必要的，以便如果用户要求跳过空事务，我们可以跳过空流，即使事务已写入一些更改。
 */
typedef struct
{
	bool		xact_wrote_changes;
	bool		stream_wrote_changes;
} TestDecodingTxnData;

static void fc_pg_decode_startup(LogicalDecodingContext *fc_ctx, OutputPluginOptions *fc_opt,
							  bool fc_is_init);
static void fc_pg_decode_shutdown(LogicalDecodingContext *fc_ctx);
static void fc_pg_decode_begin_txn(LogicalDecodingContext *fc_ctx,
								ReorderBufferTXN *fc_txn);
static void fc_pg_output_begin(LogicalDecodingContext *fc_ctx,
							TestDecodingData *fc_data,
							ReorderBufferTXN *fc_txn,
							bool fc_last_write);
static void fc_pg_decode_commit_txn(LogicalDecodingContext *fc_ctx,
								 ReorderBufferTXN *fc_txn, XLogRecPtr fc_commit_lsn);
static void fc_pg_decode_change(LogicalDecodingContext *fc_ctx,
							 ReorderBufferTXN *fc_txn, Relation fc_rel,
							 ReorderBufferChange *fc_change);
static void fc_pg_decode_truncate(LogicalDecodingContext *fc_ctx,
							   ReorderBufferTXN *fc_txn,
							   int fc_nrelations, Relation fc_relations[],
							   ReorderBufferChange *fc_change);
static bool fc_pg_decode_filter(LogicalDecodingContext *fc_ctx,
							 RepOriginId fc_origin_id);
static void fc_pg_decode_message(LogicalDecodingContext *fc_ctx,
							  ReorderBufferTXN *fc_txn, XLogRecPtr fc_message_lsn,
							  bool fc_transactional, const char *fc_prefix,
							  Size fc_sz, const char *fc_message);
static bool fc_pg_decode_filter_prepare(LogicalDecodingContext *fc_ctx,
									 TransactionId fc_xid,
									 const char *fc_gid);
static void fc_pg_decode_begin_prepare_txn(LogicalDecodingContext *fc_ctx,
										ReorderBufferTXN *fc_txn);
static void fc_pg_decode_prepare_txn(LogicalDecodingContext *fc_ctx,
								  ReorderBufferTXN *fc_txn,
								  XLogRecPtr fc_prepare_lsn);
static void fc_pg_decode_commit_prepared_txn(LogicalDecodingContext *fc_ctx,
										  ReorderBufferTXN *fc_txn,
										  XLogRecPtr fc_commit_lsn);
static void fc_pg_decode_rollback_prepared_txn(LogicalDecodingContext *fc_ctx,
											ReorderBufferTXN *fc_txn,
											XLogRecPtr fc_prepare_end_lsn,
											TimestampTz fc_prepare_time);
static void fc_pg_decode_stream_start(LogicalDecodingContext *fc_ctx,
								   ReorderBufferTXN *fc_txn);
static void fc_pg_output_stream_start(LogicalDecodingContext *fc_ctx,
								   TestDecodingData *fc_data,
								   ReorderBufferTXN *fc_txn,
								   bool fc_last_write);
static void fc_pg_decode_stream_stop(LogicalDecodingContext *fc_ctx,
								  ReorderBufferTXN *fc_txn);
static void fc_pg_decode_stream_abort(LogicalDecodingContext *fc_ctx,
								   ReorderBufferTXN *fc_txn,
								   XLogRecPtr fc_abort_lsn);
static void fc_pg_decode_stream_prepare(LogicalDecodingContext *fc_ctx,
									 ReorderBufferTXN *fc_txn,
									 XLogRecPtr fc_prepare_lsn);
static void fc_pg_decode_stream_commit(LogicalDecodingContext *fc_ctx,
									ReorderBufferTXN *fc_txn,
									XLogRecPtr fc_commit_lsn);
static void fc_pg_decode_stream_change(LogicalDecodingContext *fc_ctx,
									ReorderBufferTXN *fc_txn,
									Relation fc_relation,
									ReorderBufferChange *fc_change);
static void fc_pg_decode_stream_message(LogicalDecodingContext *fc_ctx,
									 ReorderBufferTXN *fc_txn, XLogRecPtr fc_message_lsn,
									 bool fc_transactional, const char *fc_prefix,
									 Size fc_sz, const char *fc_message);
static void fc_pg_decode_stream_truncate(LogicalDecodingContext *fc_ctx,
									  ReorderBufferTXN *fc_txn,
									  int fc_nrelations, Relation fc_relations[],
									  ReorderBufferChange *fc_change);

void _PG_init(void)
{
	/* 其他插件可以在这里执行操作 */
}

/* 指定输出插件回调 */
void _PG_output_plugin_init(OutputPluginCallbacks *fc_cb)
{
	AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit);

	fc_cb->startup_cb = fc_pg_decode_startup;
	fc_cb->begin_cb = fc_pg_decode_begin_txn;
	fc_cb->change_cb = fc_pg_decode_change;
	fc_cb->truncate_cb = fc_pg_decode_truncate;
	fc_cb->commit_cb = fc_pg_decode_commit_txn;
	fc_cb->filter_by_origin_cb = fc_pg_decode_filter;
	fc_cb->shutdown_cb = fc_pg_decode_shutdown;
	fc_cb->message_cb = fc_pg_decode_message;
	fc_cb->filter_prepare_cb = fc_pg_decode_filter_prepare;
	fc_cb->begin_prepare_cb = fc_pg_decode_begin_prepare_txn;
	fc_cb->prepare_cb = fc_pg_decode_prepare_txn;
	fc_cb->commit_prepared_cb = fc_pg_decode_commit_prepared_txn;
	fc_cb->rollback_prepared_cb = fc_pg_decode_rollback_prepared_txn;
	fc_cb->stream_start_cb = fc_pg_decode_stream_start;
	fc_cb->stream_stop_cb = fc_pg_decode_stream_stop;
	fc_cb->stream_abort_cb = fc_pg_decode_stream_abort;
	fc_cb->stream_prepare_cb = fc_pg_decode_stream_prepare;
	fc_cb->stream_commit_cb = fc_pg_decode_stream_commit;
	fc_cb->stream_change_cb = fc_pg_decode_stream_change;
	fc_cb->stream_message_cb = fc_pg_decode_stream_message;
	fc_cb->stream_truncate_cb = fc_pg_decode_stream_truncate;
}


/* 初始化此插件 */
static void fc_pg_decode_startup(LogicalDecodingContext *fc_ctx, OutputPluginOptions *fc_opt,
				  bool fc_is_init)
{
	ListCell   *fc_option;
	TestDecodingData *fc_data;
	bool		fc_enable_streaming = false;

	fc_data = palloc0(sizeof(TestDecodingData));
	fc_data->context = AllocSetContextCreate(fc_ctx->context,
										  "text conversion context",
										  ALLOCSET_DEFAULT_SIZES);
	fc_data->include_xids = true;
	fc_data->include_timestamp = false;
	fc_data->skip_empty_xacts = false;
	fc_data->only_local = false;

	fc_ctx->output_plugin_private = fc_data;

	fc_opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
	fc_opt->receive_rewrites = false;

	foreach(fc_option, fc_ctx->output_plugin_options)
	{
		DefElem    *fc_elem = lfirst(fc_option);

		Assert(fc_elem->arg == NULL || IsA(fc_elem->arg, String));

		if (strcmp(fc_elem->defname, "include-xids") == 0)
		{
			/* 如果选项未提供值，则意味着其值为真 */
			if (fc_elem->arg == NULL)
				fc_data->include_xids = true;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_data->include_xids))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else if (strcmp(fc_elem->defname, "include-timestamp") == 0)
		{
			if (fc_elem->arg == NULL)
				fc_data->include_timestamp = true;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_data->include_timestamp))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else if (strcmp(fc_elem->defname, "force-binary") == 0)
		{
			bool		fc_force_binary;

			if (fc_elem->arg == NULL)
				continue;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_force_binary))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));

			if (fc_force_binary)
				fc_opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
		}
		else if (strcmp(fc_elem->defname, "skip-empty-xacts") == 0)
		{

			if (fc_elem->arg == NULL)
				fc_data->skip_empty_xacts = true;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_data->skip_empty_xacts))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else if (strcmp(fc_elem->defname, "only-local") == 0)
		{

			if (fc_elem->arg == NULL)
				fc_data->only_local = true;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_data->only_local))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else if (strcmp(fc_elem->defname, "include-rewrites") == 0)
		{

			if (fc_elem->arg == NULL)
				continue;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_opt->receive_rewrites))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else if (strcmp(fc_elem->defname, "stream-changes") == 0)
		{
			if (fc_elem->arg == NULL)
				continue;
			else if (!parse_bool(strVal(fc_elem->arg), &fc_enable_streaming))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
								strVal(fc_elem->arg), fc_elem->defname)));
		}
		else
		{
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("option \"%s\" = \"%s\" is unknown",
							fc_elem->defname,
							fc_elem->arg ? strVal(fc_elem->arg) : "(null)")));
		}
	}

	fc_ctx->streaming &= fc_enable_streaming;
}

/* 清理此插件的资源 */
static void fc_pg_decode_shutdown(LogicalDecodingContext *fc_ctx)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;

	/* 通过内存上下文重置清理我们自己的资源 */
	MemoryContextDelete(fc_data->context);
}

/* 开始回调 */
static void fc_pg_decode_begin_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata =
	MemoryContextAllocZero(fc_ctx->context, sizeof(TestDecodingTxnData));

	fc_txndata->xact_wrote_changes = false;
	fc_txn->output_plugin_private = fc_txndata;

	/*
	 * 如果请求跳过空事务，我们将在接收到此事务的第一操作时发出 BEGIN。
	 */
	if (fc_data->skip_empty_xacts)
		return;

	fc_pg_output_begin(fc_ctx, fc_data, fc_txn, true);
}

static void fc_pg_output_begin(LogicalDecodingContext *fc_ctx, TestDecodingData *fc_data, ReorderBufferTXN *fc_txn, bool fc_last_write)
{
	OutputPluginPrepareWrite(fc_ctx, fc_last_write);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "BEGIN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "BEGIN");
	OutputPluginWrite(fc_ctx, fc_last_write);
}

/* 提交回调 */
static void fc_pg_decode_commit_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
					 XLogRecPtr fc_commit_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;
	bool		fc_xact_wrote_changes = fc_txndata->xact_wrote_changes;

	pfree(fc_txndata);
	fc_txn->output_plugin_private = NULL;

	if (fc_data->skip_empty_xacts && !fc_xact_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "COMMIT %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "COMMIT");

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.commit_time));

	OutputPluginWrite(fc_ctx, true);
}

/* 准备开始回调 */
static void fc_pg_decode_begin_prepare_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata =
	MemoryContextAllocZero(fc_ctx->context, sizeof(TestDecodingTxnData));

	fc_txndata->xact_wrote_changes = false;
	fc_txn->output_plugin_private = fc_txndata;

	/*
	 * 如果请求跳过空事务，我们将在接收到此事务的第一操作时发出 BEGIN。
	 */
	if (fc_data->skip_empty_xacts)
		return;

	fc_pg_output_begin(fc_ctx, fc_data, fc_txn, true);
}

/* 准备回调 */
static void fc_pg_decode_prepare_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
					  XLogRecPtr fc_prepare_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	/*
	 * 如果请求跳过空事务，我们将在接收到此事务的第一操作时发出 PREPARE。
	 */
	if (fc_data->skip_empty_xacts && !fc_txndata->xact_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);

	appendStringInfo(fc_ctx->out, "PREPARE TRANSACTION %s",
					 quote_literal_cstr(fc_txn->gid));

	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, ", txid %u", fc_txn->xid);

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.prepare_time));

	OutputPluginWrite(fc_ctx, true);
}

/* 提交已准备好的回调 */
static void fc_pg_decode_commit_prepared_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
							  XLogRecPtr fc_commit_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;

	OutputPluginPrepareWrite(fc_ctx, true);

	appendStringInfo(fc_ctx->out, "COMMIT PREPARED %s",
					 quote_literal_cstr(fc_txn->gid));

	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, ", txid %u", fc_txn->xid);

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.commit_time));

	OutputPluginWrite(fc_ctx, true);
}

/* 回滚已准备好的回调 */
static void fc_pg_decode_rollback_prepared_txn(LogicalDecodingContext *fc_ctx,
								ReorderBufferTXN *fc_txn,
								XLogRecPtr fc_prepare_end_lsn,
								TimestampTz fc_prepare_time)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;

	OutputPluginPrepareWrite(fc_ctx, true);

	appendStringInfo(fc_ctx->out, "ROLLBACK PREPARED %s",
					 quote_literal_cstr(fc_txn->gid));

	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, ", txid %u", fc_txn->xid);

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.commit_time));

	OutputPluginWrite(fc_ctx, true);
}

/*
 * 过滤掉两阶段事务。
 *
 * 每个插件可以实现自己的过滤逻辑。这里我们通过检查 GID 演示一个简单的逻辑。 如果 GID 包含 "_nodecode" 子字符串，则我们将其过滤掉。
 */
static bool fc_pg_decode_filter_prepare(LogicalDecodingContext *fc_ctx, TransactionId fc_xid,
						 const char *fc_gid)
{
	if (strstr(fc_gid, "_nodecode") != NULL)
		return true;

	return false;
}

static bool fc_pg_decode_filter(LogicalDecodingContext *fc_ctx,
				 RepOriginId fc_origin_id)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;

	if (fc_data->only_local && fc_origin_id != InvalidRepOriginId)
		return true;
	return false;
}

/*
 * 将字面量 `outputstr' 作为 `typid` 类型的字符串打印到字符串缓冲区 `s` 中。
 *
 * 一些内置类型不加引号，其余的则加引号。 转义按标准符合字符串启用的方式进行。
 */
static void fc_print_literal(StringInfo fc_s, Oid fc_typid, char *fc_outputstr)
{
	const char *fc_valptr;

	switch (fc_typid)
	{
		case INT2OID:
		case INT4OID:
		case INT8OID:
		case OIDOID:
		case FLOAT4OID:
		case FLOAT8OID:
		case NUMERICOID:
			/* 注意：我们不关心 Inf、NaN 等。 */
			appendStringInfoString(fc_s, fc_outputstr);
			break;

		case BITOID:
		case VARBITOID:
			appendStringInfo(fc_s, "B'%s'", fc_outputstr);
			break;

		case BOOLOID:
			if (strcmp(fc_outputstr, "t") == 0)
				appendStringInfoString(fc_s, "true");
			else
				appendStringInfoString(fc_s, "false");
			break;

		default:
			appendStringInfoChar(fc_s, '\'');
			for (fc_valptr = fc_outputstr; *fc_valptr; fc_valptr++)
			{
				char		fc_ch = *fc_valptr;

				if (SQL_STR_DOUBLE(fc_ch, false))
					appendStringInfoChar(fc_s, fc_ch);
				appendStringInfoChar(fc_s, fc_ch);
			}
			appendStringInfoChar(fc_s, '\'');
			break;
	}
}

/* 将元组 'tuple' 打印到字符串信息 s 中 */
static void fc_tuple_to_stringinfo(StringInfo fc_s, TupleDesc fc_tupdesc, HeapTuple fc_tuple, bool fc_skip_nulls)
{
	int			fc_natt;

	/* 单独打印所有列 */
	for (fc_natt = 0; fc_natt < fc_tupdesc->natts; fc_natt++)
	{
		Form_pg_attribute fc_attr; /* 属性本身 */
		Oid			fc_typid;		/* 当前属性的类型 */
		Oid			fc_typoutput;	/* 输出函数 */
		bool		fc_typisvarlena;
		Datum		fc_origval;	/* 可能被烘烤的 Datum */
		bool		fc_isnull;		/* 列是否为 null？ */

		fc_attr = TupleDescAttr(fc_tupdesc, fc_natt);

		/*
		 * 不打印被丢弃的列，我们不能确定对它们是否一切可用
		 */
		if (fc_attr->attisdropped)
			continue;

		/*
		 * 不打印系统列，如果存在，oid 已经被打印出来。
		 */
		if (fc_attr->attnum < 0)
			continue;

		fc_typid = fc_attr->atttypid;

		/* 从元组中获取 Datum */
		fc_origval = heap_getattr(fc_tuple, fc_natt + 1, fc_tupdesc, &fc_isnull);

		if (fc_isnull && fc_skip_nulls)
			continue;

		/* 打印属性名称 */
		appendStringInfoChar(fc_s, ' ');
		appendStringInfoString(fc_s, quote_identifier(NameStr(fc_attr->attname)));

		/* 打印属性类型 */
		appendStringInfoChar(fc_s, '[');
		appendStringInfoString(fc_s, format_type_be(fc_typid));
		appendStringInfoChar(fc_s, ']');

		/* 查询输出函数 */
		getTypeOutputInfo(fc_typid,
						  &fc_typoutput, &fc_typisvarlena);

		/* 打印分隔符 */
		appendStringInfoChar(fc_s, ':');

		/* 打印数据 */
		if (fc_isnull)
			appendStringInfoString(fc_s, "null");
		else if (fc_typisvarlena && VARATT_IS_EXTERNAL_ONDISK(fc_origval))
			appendStringInfoString(fc_s, "unchanged-toast-datum");
		else if (!fc_typisvarlena)
			fc_print_literal(fc_s, fc_typid,
						  OidOutputFunctionCall(fc_typoutput, fc_origval));
		else
		{
			Datum		fc_val;	/* 绝对解烘烤的 Datum */

			fc_val = PointerGetDatum(PG_DETOAST_DATUM(fc_origval));
			fc_print_literal(fc_s, fc_typid, OidOutputFunctionCall(fc_typoutput, fc_val));
		}
	}
}

/*
 * 对单个已更改元组的回调
 */
static void fc_pg_decode_change(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
				 Relation fc_relation, ReorderBufferChange *fc_change)
{
	TestDecodingData *fc_data;
	TestDecodingTxnData *fc_txndata;
	Form_pg_class fc_class_form;
	TupleDesc	fc_tupdesc;
	MemoryContext fc_old;

	fc_data = fc_ctx->output_plugin_private;
	fc_txndata = fc_txn->output_plugin_private;

	/* 如果我们还没有，输出 BEGIN */
	if (fc_data->skip_empty_xacts && !fc_txndata->xact_wrote_changes)
	{
		fc_pg_output_begin(fc_ctx, fc_data, fc_txn, false);
	}
	fc_txndata->xact_wrote_changes = true;

	fc_class_form = RelationGetForm(fc_relation);
	fc_tupdesc = RelationGetDescr(fc_relation);

	/* 通过使用并重置我们自己的上下文来避免内存泄漏 */
	fc_old = MemoryContextSwitchTo(fc_data->context);

	OutputPluginPrepareWrite(fc_ctx, true);

	appendStringInfoString(fc_ctx->out, "table ");
	appendStringInfoString(fc_ctx->out,
						   quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(fc_relation))),
													  fc_class_form->relrewrite ?
													  get_rel_name(fc_class_form->relrewrite) :
													  NameStr(fc_class_form->relname)));
	appendStringInfoChar(fc_ctx->out, ':');

	switch (fc_change->action)
	{
		case REORDER_BUFFER_CHANGE_INSERT:
			appendStringInfoString(fc_ctx->out, " INSERT:");
			if (fc_change->data.tp.newtuple == NULL)
				appendStringInfoString(fc_ctx->out, " (no-tuple-data)");
			else
				fc_tuple_to_stringinfo(fc_ctx->out, fc_tupdesc,
									&fc_change->data.tp.newtuple->tuple,
									false);
			break;
		case REORDER_BUFFER_CHANGE_UPDATE:
			appendStringInfoString(fc_ctx->out, " UPDATE:");
			if (fc_change->data.tp.oldtuple != NULL)
			{
				appendStringInfoString(fc_ctx->out, " old-key:");
				fc_tuple_to_stringinfo(fc_ctx->out, fc_tupdesc,
									&fc_change->data.tp.oldtuple->tuple,
									true);
				appendStringInfoString(fc_ctx->out, " new-tuple:");
			}

			if (fc_change->data.tp.newtuple == NULL)
				appendStringInfoString(fc_ctx->out, " (no-tuple-data)");
			else
				fc_tuple_to_stringinfo(fc_ctx->out, fc_tupdesc,
									&fc_change->data.tp.newtuple->tuple,
									false);
			break;
		case REORDER_BUFFER_CHANGE_DELETE:
			appendStringInfoString(fc_ctx->out, " DELETE:");

			/* 如果没有主键，我们只知道发生了一次删除 */
			if (fc_change->data.tp.oldtuple == NULL)
				appendStringInfoString(fc_ctx->out, " (no-tuple-data)");
			/* 在删除操作中，仅存在副本标识；显示该内容 */
			else
				fc_tuple_to_stringinfo(fc_ctx->out, fc_tupdesc,
									&fc_change->data.tp.oldtuple->tuple,
									true);
			break;
		default:
			Assert(false);
	}

	MemoryContextSwitchTo(fc_old);
	MemoryContextReset(fc_data->context);

	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_truncate(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
				   int fc_nrelations, Relation fc_relations[], ReorderBufferChange *fc_change)
{
	TestDecodingData *fc_data;
	TestDecodingTxnData *fc_txndata;
	MemoryContext fc_old;
	int			fc_i;

	fc_data = fc_ctx->output_plugin_private;
	fc_txndata = fc_txn->output_plugin_private;

	/* 如果我们还没有，输出 BEGIN */
	if (fc_data->skip_empty_xacts && !fc_txndata->xact_wrote_changes)
	{
		fc_pg_output_begin(fc_ctx, fc_data, fc_txn, false);
	}
	fc_txndata->xact_wrote_changes = true;

	/* 通过使用并重置我们自己的上下文来避免内存泄漏 */
	fc_old = MemoryContextSwitchTo(fc_data->context);

	OutputPluginPrepareWrite(fc_ctx, true);

	appendStringInfoString(fc_ctx->out, "table ");

	for (fc_i = 0; fc_i < fc_nrelations; fc_i++)
	{
		if (fc_i > 0)
			appendStringInfoString(fc_ctx->out, ", ");

		appendStringInfoString(fc_ctx->out,
							   quote_qualified_identifier(get_namespace_name(fc_relations[fc_i]->rd_rel->relnamespace),
														  NameStr(fc_relations[fc_i]->rd_rel->relname)));
	}

	appendStringInfoString(fc_ctx->out, ": TRUNCATE:");

	if (fc_change->data.truncate.restart_seqs
		|| fc_change->data.truncate.cascade)
	{
		if (fc_change->data.truncate.restart_seqs)
			appendStringInfoString(fc_ctx->out, " restart_seqs");
		if (fc_change->data.truncate.cascade)
			appendStringInfoString(fc_ctx->out, " cascade");
	}
	else
		appendStringInfoString(fc_ctx->out, " (no-flags)");

	MemoryContextSwitchTo(fc_old);
	MemoryContextReset(fc_data->context);

	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_message(LogicalDecodingContext *fc_ctx,
				  ReorderBufferTXN *fc_txn, XLogRecPtr fc_lsn, bool fc_transactional,
				  const char *fc_prefix, Size fc_sz, const char *fc_message)
{
	OutputPluginPrepareWrite(fc_ctx, true);
	appendStringInfo(fc_ctx->out, "message: transactional: %d prefix: %s, sz: %zu content:",
					 fc_transactional, fc_prefix, fc_sz);
	appendBinaryStringInfo(fc_ctx->out, fc_message, fc_sz);
	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_stream_start(LogicalDecodingContext *fc_ctx,
					   ReorderBufferTXN *fc_txn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	/*
	 * 为事务中的第一个流分配事务插件数据。
	 */
	if (fc_txndata == NULL)
	{
		fc_txndata =
			MemoryContextAllocZero(fc_ctx->context, sizeof(TestDecodingTxnData));
		fc_txndata->xact_wrote_changes = false;
		fc_txn->output_plugin_private = fc_txndata;
	}

	fc_txndata->stream_wrote_changes = false;
	if (fc_data->skip_empty_xacts)
		return;
	fc_pg_output_stream_start(fc_ctx, fc_data, fc_txn, true);
}

static void fc_pg_output_stream_start(LogicalDecodingContext *fc_ctx, TestDecodingData *fc_data, ReorderBufferTXN *fc_txn, bool fc_last_write)
{
	OutputPluginPrepareWrite(fc_ctx, fc_last_write);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "opening a streamed block for transaction TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "opening a streamed block for transaction");
	OutputPluginWrite(fc_ctx, fc_last_write);
}

static void fc_pg_decode_stream_stop(LogicalDecodingContext *fc_ctx,
					  ReorderBufferTXN *fc_txn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	if (fc_data->skip_empty_xacts && !fc_txndata->stream_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "closing a streamed block for transaction TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "closing a streamed block for transaction");
	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_stream_abort(LogicalDecodingContext *fc_ctx,
					   ReorderBufferTXN *fc_txn,
					   XLogRecPtr fc_abort_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;

	/*
	 * 流式中止可以针对单个子事务发送，但我们
	 * 仅在顶级事务下维护 output_plugin_private，因此如果这不是
	 * 顶级事务，则获取顶级事务。
	 */
	ReorderBufferTXN *fc_toptxn = fc_txn->toptxn ? fc_txn->toptxn : fc_txn;
	TestDecodingTxnData *fc_txndata = fc_toptxn->output_plugin_private;
	bool		fc_xact_wrote_changes = fc_txndata->xact_wrote_changes;

	if (fc_txn->toptxn == NULL)
	{
		Assert(fc_txn->output_plugin_private != NULL);
		pfree(fc_txndata);
		fc_txn->output_plugin_private = NULL;
	}

	if (fc_data->skip_empty_xacts && !fc_xact_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "aborting streamed (sub)transaction TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "aborting streamed (sub)transaction");
	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_stream_prepare(LogicalDecodingContext *fc_ctx,
						 ReorderBufferTXN *fc_txn,
						 XLogRecPtr fc_prepare_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	if (fc_data->skip_empty_xacts && !fc_txndata->xact_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);

	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "preparing streamed transaction TXN %s, txid %u",
						 quote_literal_cstr(fc_txn->gid), fc_txn->xid);
	else
		appendStringInfo(fc_ctx->out, "preparing streamed transaction %s",
						 quote_literal_cstr(fc_txn->gid));

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.prepare_time));

	OutputPluginWrite(fc_ctx, true);
}

static void fc_pg_decode_stream_commit(LogicalDecodingContext *fc_ctx,
						ReorderBufferTXN *fc_txn,
						XLogRecPtr fc_commit_lsn)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;
	bool		fc_xact_wrote_changes = fc_txndata->xact_wrote_changes;

	pfree(fc_txndata);
	fc_txn->output_plugin_private = NULL;

	if (fc_data->skip_empty_xacts && !fc_xact_wrote_changes)
		return;

	OutputPluginPrepareWrite(fc_ctx, true);

	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "committing streamed transaction TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "committing streamed transaction");

	if (fc_data->include_timestamp)
		appendStringInfo(fc_ctx->out, " (at %s)",
						 timestamptz_to_str(fc_txn->xact_time.commit_time));

	OutputPluginWrite(fc_ctx, true);
}

/*
 * 在流式模式下，我们不显示变更，因为事务可能在稍后的时间点中止。
 * 我们不希望用户在事务提交之前看到变更。
 */
static void fc_pg_decode_stream_change(LogicalDecodingContext *fc_ctx,
						ReorderBufferTXN *fc_txn,
						Relation fc_relation,
						ReorderBufferChange *fc_change)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	/* 如果我们尚未输出，则输出流开始 */
	if (fc_data->skip_empty_xacts && !fc_txndata->stream_wrote_changes)
	{
		fc_pg_output_stream_start(fc_ctx, fc_data, fc_txn, false);
	}
	fc_txndata->xact_wrote_changes = fc_txndata->stream_wrote_changes = true;

	OutputPluginPrepareWrite(fc_ctx, true);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "streaming change for TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "streaming change for transaction");
	OutputPluginWrite(fc_ctx, true);
}

/*
 * 在流式模式下，我们不显示交易消息的内容，
 * 因为事务可能在稍后时间点中止。我们不希望用户在
 * 事务提交之前看到消息内容。
 */
static void fc_pg_decode_stream_message(LogicalDecodingContext *fc_ctx,
						 ReorderBufferTXN *fc_txn, XLogRecPtr fc_lsn, bool fc_transactional,
						 const char *fc_prefix, Size fc_sz, const char *fc_message)
{
	OutputPluginPrepareWrite(fc_ctx, true);

	if (fc_transactional)
	{
		appendStringInfo(fc_ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu",
						 fc_transactional, fc_prefix, fc_sz);
	}
	else
	{
		appendStringInfo(fc_ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu content:",
						 fc_transactional, fc_prefix, fc_sz);
		appendBinaryStringInfo(fc_ctx->out, fc_message, fc_sz);
	}

	OutputPluginWrite(fc_ctx, true);
}

/*
 * 在流式模式下，我们不显示 Truncate 的详细信息。
 * 请参阅 pg_decode_stream_change。
 */
static void fc_pg_decode_stream_truncate(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
						  int fc_nrelations, Relation fc_relations[],
						  ReorderBufferChange *fc_change)
{
	TestDecodingData *fc_data = fc_ctx->output_plugin_private;
	TestDecodingTxnData *fc_txndata = fc_txn->output_plugin_private;

	if (fc_data->skip_empty_xacts && !fc_txndata->stream_wrote_changes)
	{
		fc_pg_output_stream_start(fc_ctx, fc_data, fc_txn, false);
	}
	fc_txndata->xact_wrote_changes = fc_txndata->stream_wrote_changes = true;

	OutputPluginPrepareWrite(fc_ctx, true);
	if (fc_data->include_xids)
		appendStringInfo(fc_ctx->out, "streaming truncate for TXN %u", fc_txn->xid);
	else
		appendStringInfoString(fc_ctx->out, "streaming truncate for transaction");
	OutputPluginWrite(fc_ctx, true);
}
